package com.errol.batchprocessor.batchprocessor.advisorimpl.timeconsole;

import com.renjia.entity.AbstractImportBatch;
import com.renjia.service.generic.base.util.DateFormatUtil;
import com.renjia.service.generic.log.LogService;
import com.renjia.service.generic.middleware.batchprocessor.advisor.*;
import com.renjia.service.generic.middleware.batchprocessor.bo.ThreadPoolStatusBO;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractBatchPayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractPiecePayload;
import com.renjia.service.generic.middleware.batchprocessor.payload.AbstractSinglePayload;
import org.springframework.lang.NonNull;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 批处理执行时间统计的扩展，实现描述如下：
 * - 批次加入队列时：记录添加的时间、输出提示
 * - 批次启动时：记录启动时间、并输出等待时长
 * - 分片启动时：记录启动时间、并输出提示第几个分片开始执行
 * - 分片结算后：计算分片用时、输出提示及分片用时
 * - 批次结算后：输出批次总用时、分片总数及成功完成数据、分片平均用时
 *
 * @author zhang xiao bin
 * @version v1.0
 * @date 2021/03/12
 * @since v5.0.6
 **/
public abstract class AbstractTimeConsoleAdvisorImpl<IE extends AbstractImportBatch, AT extends Enum<?>, BE, SE, BP extends AbstractBatchPayload<IE, AT, BE, SE>, PP extends AbstractPiecePayload<SE, SP>, SP extends AbstractSinglePayload<SE>>
        implements
        BatchAddAdvisor<IE, AT, BE, SE, BP, PP, SP>, BatchStartAdvisor<IE, AT, BE, SE, BP, PP, SP>, BatchPostAdvisor<IE, AT, BE, SE, BP, PP, SP>,
        PieceStartAdvisor<IE, AT, BE, SE, BP, PP, SP>, PiecePostAdvisor<IE, AT, BE, SE, BP, PP, SP> {

    private final ConcurrentMap<String, TimeConsole> timeMap;
    protected final LogService logService;

    public AbstractTimeConsoleAdvisorImpl(LogService logService) {
        this.logService = logService;
        this.timeMap = new ConcurrentHashMap<>();
    }

    @Override
    public int priority() {
        return 1;
    }

    @Override
    public List<PP> doSplitOnBatchAdded(@NonNull BP batchPayload, @NonNull ThreadPoolStatusBO threadPoolStatus) {
        if (batchPayload.getImportBatch().getId() != null) {
            timeMap.putIfAbsent(batchPayload.getBatchNo(), new TimeConsole());
            logService.debug(logPrefix() + wrapBatchNo(batchPayload) + " 加入批处理队列");
        }
        return null;
    }

    @Override
    public boolean doPrepareOnBatchStart(@NonNull BP batchPayload) {
        TimeConsole timeConsole = timeMap.get(batchPayload.getBatchNo());
        if (timeConsole != null) {
            timeConsole.batchStartTime = System.currentTimeMillis();
            logService.debug(logPrefix() + wrapBatchNo(batchPayload) + " 批次开始执行，等待时长：" + formatTime(timeConsole.batchStartTime - timeConsole.batchAddTime));
        }
        return true;
    }

    @Override
    public boolean doCleanUpOnBatchPost(@NonNull BP batchPayload) {
        TimeConsole timeConsole = timeMap.remove(batchPayload.getBatchNo());
        if (timeConsole != null) {
            if (timeConsole.pieceFinishCount > 0) {
                timeConsole.batchFinishTime = System.currentTimeMillis();
                logService.debug(logPrefix() + wrapBatchNo(batchPayload) + " 批次执行完成，总计用时：" + formatTime(timeConsole.batchFinishTime - timeConsole.batchStartTime)
                        + " | 批次分片：总计 " + timeConsole.pieceStartTimeMap.size() + " 完成 " + timeConsole.pieceFinishCount
                        + " 平均用时 " + formatTime(timeConsole.pieceUsedTimeAmount / timeConsole.pieceFinishCount));
            } else {
                logService.debug(logPrefix() + wrapBatchNo(batchPayload) + " 批次执行完成：批次启动失败或所有分片全部执行异常");
            }
        }
        return true;
    }

    @Override
    public boolean doPrepareOnPieceStart(@NonNull PP piecePayload, @NonNull BP batchPayload) {
        TimeConsole timeConsole = timeMap.get(batchPayload.getBatchNo());
        if (timeConsole != null) {
            timeConsole.pieceStartTimeMap.putIfAbsent(piecePayload, System.currentTimeMillis());
            logService.debug(logPrefix() + wrapBatchNo(batchPayload) + " 开始执行第 " + timeConsole.pieceStartTimeMap.size()
                    + " 个分片，分片内数据量：" + piecePayload.getSinglePayloadList().size());
        }
        return true;
    }

    @Override
    public boolean doCleanUpOnPiecePost(@NonNull PP piecePayload, @NonNull BP batchPayload) {
        TimeConsole timeConsole = timeMap.get(batchPayload.getBatchNo());
        if (timeConsole != null) {
            Long time = timeConsole.pieceStartTimeMap.get(piecePayload);
            if (time != null) {
                time = System.currentTimeMillis() - time;
                timeConsole.pieceFinishCount++;
                timeConsole.pieceUsedTimeAmount += time;
                logService.debug(logPrefix() + wrapBatchNo(batchPayload) + " 已完成 " + timeConsole.pieceFinishCount + " 个分片，当前分片用时：" + formatTime(time));
            }
        }
        return true;
    }

    /**
     * 输出日志时的前缀
     *
     * @return 输出日志时的前缀
     */
    protected abstract String logPrefix();

    /**
     * 包装批次号的输出格式
     *
     * @param batchPayload bp
     * @return 批次号格式化输出
     */
    protected String wrapBatchNo(BP batchPayload) {
        return "【" + batchPayload.getBatchNo() + "】";
    }

    /**
     * 格式化用时
     *
     * @param ms 毫秒
     * @return HH:mm:ss_SSS
     */
    protected String formatTime(long ms) {
        String s = DateFormatUtil.stamp2String(ms - 28800000, DateFormatUtil.PATTERN_S);
        if (s == null) {
            return ms + "ms";
        }
        return s;
    }
}
